package com.cqwshzj.thread.worker;/**
 * @Description:
 * @Auther: Huang Zhijin
 * @Version: 1.0.0
 * @Date: 2021/4/12 10:41
 */

import com.cqwshzj.common.JobConstant;
import com.cqwshzj.thread.bean.Stateable;
import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 *
 * 〈〉
 *
 * @author Huang Zhijin
 * @create 2021/4/12
 * @since 1.0.0
 */
public class MainBizWorker<T extends Stateable> implements Runnable{

	private static final Logger log = LoggerFactory.getLogger(MainBizWorker.class);

	//------------------------------

	private LinkedBlockingQueue<T> downloadQueue;
	private LinkedBlockingQueue<T>  checkQueue;

	private CountDownLatch mainWorkerSignal;
	private CountDownLatch filesCountSignal;
	private CountDownLatch assistantThreadSignal;


	//------------------------------
	/** 业务次数 */
	private int bizCount = 0;

	//---------------------

	public MainBizWorker(LinkedBlockingQueue<T> downloadQueue,
						 LinkedBlockingQueue<T> checkQueue,
						 CountDownLatch mainWorkerSignal,
						 CountDownLatch filesCountSignal,
						 CountDownLatch assistantThreadSignal) {
		this.downloadQueue = downloadQueue;
		this.checkQueue = checkQueue;
		this.mainWorkerSignal = mainWorkerSignal;
		this.filesCountSignal = filesCountSignal;
		this.assistantThreadSignal = assistantThreadSignal;
	}


	//---------------------

	@Override
	public void run() {
		log.info("--------  main biz worker start   --------");
		long start = System.currentTimeMillis();
		int count = 0;
		long waitedTime = 0;
		try {
			while (0 != downloadQueue.size() + checkQueue.size() + assistantThreadSignal.getCount() ){
				if ( 0 == downloadQueue.size() ) {
					try {
						int sleepPeriod = 50;
						Thread.sleep( sleepPeriod );
						waitedTime += sleepPeriod;
						if ( waitedTime % (sleepPeriod*100) == 0) {
							log.debug(" downloadQueue=[{}], checkQueue=[{}], assistantThreadSignalCount=[{}]. ", downloadQueue.size(), checkQueue.size(), assistantThreadSignal.getCount());
						}
					}catch (InterruptedException e){
						log.error(e.getMessage(),e);
						return;
					}
					continue;
				}

				downloadSubject( downloadQueue.peek() );
			}
		}catch (Throwable e){
			log.error(e.getMessage(),e);
		}
		log.info(" download OK! count=[{}], cost=[\t{}]ms, thread sleep=[\t{}]ms.",count, (System.currentTimeMillis()-start), waitedTime);
		mainWorkerSignal.countDown();
		log.info("--------  main biz worker end   --------");
	}


	private void downloadSubject(T subject) {
		bizCount++;
		log.info(" download subject . id=[{}] state=[{}] .", subject.getIdentityField(), subject.getState());
		try {
			final int timeout = RandomUtils.nextInt(1, 5);
			if (timeout > 1 ){
				subject.setState( JobConstant.COMMON_YES );
			}else{
				subject.setState( JobConstant.COMMON_NO );
			}
			TimeUnit.MILLISECONDS.sleep(timeout);

			checkQueue.offer( subject );
			downloadQueue.remove( subject );
			return;
		} catch (Exception e) {
			log.error(e.getMessage(),e);
			downloadQueue.remove( subject );
			//重新下载
			downloadQueue.offer( subject );
		}

	}


}
